/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


package com.sui.bigdata.flink.sql.core.side.operator;

import com.sui.bigdata.flink.sql.core.side.*;
import com.sui.bigdata.flink.sql.core.side.AsyncReqRow;
import com.sui.bigdata.flink.sql.core.side.FieldInfo;
import com.sui.bigdata.flink.sql.core.side.JoinInfo;
import com.sui.bigdata.flink.sql.core.side.SideTableInfo;
import com.sui.bigdata.flink.sql.side.hbase.HbaseAsyncReqRow;
import com.sui.bigdata.flink.sql.side.mysql.MysqlAsyncReqRow;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import com.sui.bigdata.flink.sql.side.kudu.async.KuduAsyncReqRow;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * fill data with lru cache
 * get data from  External data source with async operator
 * Date: 2019/07/10
 * @author yong_chen@sui.com
 */

public class SideAsyncOperator {

    private static final String PATH_FORMAT = "%sasyncside";

    private static final String HBASE = "hbase";

    private static final String MYSQL = "mysql";

    private static final String KUDU ="kudu";
    //TODO need to set by create table task
    private static int asyncCapacity = 100;

    private static AsyncReqRow loadAsyncReq(String sideType, RowTypeInfo rowTypeInfo,
                                            JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
        if (HBASE.equals(sideType)){
            return new HbaseAsyncReqRow(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
        }

        if (MYSQL.equals(sideType)){
            return new MysqlAsyncReqRow(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
        }

        if (KUDU.equals(sideType)){
            return new KuduAsyncReqRow(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);

        }
        return null;
    }


    public static DataStream getSideJoinDataStream(DataStream inputStream, String sideType, RowTypeInfo rowTypeInfo, JoinInfo joinInfo,
                                                   List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
        AsyncReqRow asyncDbReq = loadAsyncReq(sideType, rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
        //TODO How much should be set for the degree of parallelism? Timeout? capacity settings?

       return "unorderedWait".equals(sideTableInfo.getAsyncMode())? AsyncDataStream.unorderedWait(inputStream,asyncDbReq,sideTableInfo.getAsyncTimeout(), TimeUnit.MILLISECONDS, sideTableInfo.getAsyncCapacity())
               .setParallelism(sideTableInfo.getParallelism()): AsyncDataStream.orderedWait(inputStream, asyncDbReq, sideTableInfo.getAsyncTimeout(), TimeUnit.MILLISECONDS, sideTableInfo.getAsyncCapacity())
               .setParallelism(sideTableInfo.getParallelism());
    }
}

